package com.bigbird.nio;

import io.github.bigbird0101.pipeline.*;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedTransferQueue;

public class NioServerV5 {
    private static Work[] works;
    private static final Queue<Runnable> RUNNABLE_LINKED_TRANSFER_QUEUE = new LinkedTransferQueue<Runnable>();

    public static void main(String[] args) throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(1881));
        Pipeline pipeline=new DefaultEventPipeline();
        pipeline.addFirst("string", new AbstractPredicateEventHandler(s->(s instanceof StringMessageHandlerEvent)) {
            @Override
            protected HandlerEvent doHandle(HandlerContext handlerContext, HandlerEvent handlerEvent) {
                Object source = handlerEvent.getSource();
                System.out.println("message source "+source);
                return handlerEvent;
            }
        });
        int workCount=1;
        works=new Work[workCount];
        for(int a=0;a<workCount;a++){
            works[a]= new Work(pipeline);
        }
        for(int a=0;a<workCount;a++){
            Work work = works[a];
            new Thread(work).start();
        }
        new Thread(new Boss(serverSocketChannel)).start();
    }
    public static class Boss implements Runnable{
        private final ServerSocketChannel serverSocketChannel;
        public final Selector selector;

        public Boss(ServerSocketChannel serverSocketChannel) {
            this.serverSocketChannel = serverSocketChannel;
            try {
                this.serverSocketChannel.configureBlocking(false);
                this.selector=Selector.open();
                serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void run() {
             try {
                 while(true) {
                     try {
                         SocketChannel accept = serverSocketChannel.accept();
                         if (null != accept) {
                            getNextWork().register(accept);
                         }
                     }catch (IOException exception){
                         throw new RuntimeException(exception);
                     }
                 }
             } finally {
                 try {
                     selector.close();
                 } catch (IOException e) {
                     throw new RuntimeException(e);
                 }
             }
        }

        public Work getNextWork(){
            return works[0];
        }
    }

    public static class Work implements Runnable{
        private final Selector selector;

        private final Pipeline pipeline;

        public Work(Pipeline pipeline) {
            try {
                this.pipeline=pipeline;
                this.selector=Selector.open();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        public void register(SocketChannel socketChannel){
            try {
                if(socketChannel.isConnected()) {
                    RUNNABLE_LINKED_TRANSFER_QUEUE.offer(()->{
                        try {
                            socketChannel.configureBlocking(false);
                            socketChannel.register(this.selector, SelectionKey.OP_READ);
                            this.selector.wakeup();
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    });
                }
            } catch (Exception e) {
                try {
                    socketChannel.close();
                } catch (IOException ex) {
                }
            }
        }

        @Override
        public void run() {
            try{
                while(true){
                    try {
                        selector.select(500);
                    }catch (CancelledKeyException exception){
                        exception.printStackTrace();
                    }
                    processReadKey();
                    processSelectionKey(selector.selectedKeys());
                }
            }catch (Exception exception){
                exception.printStackTrace();
            }
        }

        private void processReadKey() {
            while(true) {
                Runnable poll = RUNNABLE_LINKED_TRANSFER_QUEUE.poll();
                if (poll == null) {
                    break;
                }
                poll.run();
            }
        }

        private void processSelectionKey(Set<SelectionKey> keys) {
            Iterator<SelectionKey> iterator = keys.iterator();
            while (iterator.hasNext()) {
                SelectionKey selectionKey = iterator.next();
                //清楚已处理的
                keys.remove(selectionKey);
                if (selectionKey.interestOps() == SelectionKey.OP_READ) {
                    try {
                        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                        if(selectionKey.isValid()&&selectionKey.isReadable()) {
                            ByteBuffer byteBuffer = ByteBuffer.allocate(10);
                            int ret;
                            int readbytes=0;
                            while((ret=socketChannel.read(byteBuffer))>0) {
                                readbytes+=ret;
                                if(!byteBuffer.hasRemaining()){
                                    break;
                                }
                            }
                            if(readbytes>0){
                                byteBuffer.flip();
                                byte[] bytes = new byte[readbytes];
                                byteBuffer.get(bytes);
                                String string = new String(bytes);
                                pipeline.getFirst().ifPresent(s-> pipeline.getContext(s)
                                        .ifPresent(b-> ((EventHandler)s).handle(b,new StringMessageHandlerEvent(string))));
                            }
                            byteBuffer.rewind();
                            if(ret<0){
                                ByteBuffer byteBufferWrite = ByteBuffer.wrap("good bye".getBytes(StandardCharsets.UTF_8));
                                socketChannel.write(byteBufferWrite);
                                selectionKey.cancel();
                                socketChannel.socket().close();
                            }else {
                                ByteBuffer byteBufferWrite = ByteBuffer.wrap("good job".getBytes(StandardCharsets.UTF_8));
                                socketChannel.write(byteBufferWrite);
                            }
                            try{

                            }finally {
                                byteBuffer.clear();
                            }
                        }else{
                            selectionKey.cancel();
                        }
                    }catch (Exception exception){
                        exception.printStackTrace();
                    }
                }
                else {
                    SelectableChannel channel = selectionKey.channel();
                    System.out.println(channel);
                    System.out.println(selectionKey.interestOps());
                }
            }
        }
    }

    public static class StringMessageHandlerEvent extends HandlerEvent {
        public StringMessageHandlerEvent(String source) {
            super(source);
        }
    }
}
